package org.jxkj.data.hbase;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.jxkj.bean.PointData;
import org.jxkj.data.BaseDaoProvider;
import org.jxkj.data.IDaoProvider;
import org.jxkj.util.PropertiesUtils;
import org.jxkj.utils.HBaseUtils;
import org.jxkj.utils.ParseUtils;

import java.io.IOException;
import java.util.List;
import java.util.Properties;

/**
 * 查询宽表
 */
public class HBaseDaoWideImpl extends BaseDaoProvider implements IDaoProvider {

    private static Configuration conf;
    private static Connection connection;
    static Properties properties = PropertiesUtils.initial("hbase.properties");


    static {
        try {
            conf = HBaseConfiguration.create();
            conf.set("hbase.zookeeper.quorum", properties.getProperty("hbase.zookeeper.quorum"));
            conf.set("zookeeper.znode.parent", properties.getProperty("zookeeper.znode.parent"));
            connection = ConnectionFactory.createConnection(conf);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static Table getTable(String time) throws IOException {
        //建立连接
        String year = time.substring(0, 4);
        String tName = PropertiesUtils.getProperty("table_" + year, properties);
        // String tName = PropertiesUtils.getProperty("queryTable", properties);
        Table table = connection.getTable(TableName.valueOf(tName));
        return table;
    }


    public static Connection getConnection() {
        return connection;
    }

    /**
     * rowkey:00|015450|2019092000
     * 设置时间范围查询
     *
     * @param pointName
     * @param startTime
     * @param endTime
     * @return
     */
    public List<PointData> getByRegion(String pointName, String startTime, String endTime) throws Exception {
        Table table = getTable(endTime);
        String[] codeAndId = HBaseUtils.getCodeAndId(pointName);
        String start = ParseUtils.parseTime(startTime).substring(0, 10);
        String end = ParseUtils.parseTime(endTime).substring(0, 10);
        //生成rowkey的查询条件
        String startRowkey = codeAndId[0] + "|" + codeAndId[1] + "|" + start;
        System.out.print("startRowkey" + startRowkey);
        String endRowkey = codeAndId[0] + "|" + codeAndId[1] + "|" + end;
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRowkey));//范围查询
        scan.setStopRow(Bytes.toBytes(endRowkey));
        scan.setCacheBlocks(false);
        //scan.setCaching(1000);
        List<PointData> list = HBaseUtils.genData(table, scan, pointName);
        table.close();
        return list;
    }

    public List<PointData> getByRegionByDay(String pointName, String startTime, String endTime) throws Exception {
        Table table = getTable(endTime);
        String[] codeAndId = HBaseUtils.getCodeAndId(pointName);
        String start = ParseUtils.parseTime(startTime).substring(0, 10);
        String end = ParseUtils.parseTime(endTime);
        //生成rowkey的查询条件
        String startRowkey = codeAndId[0] + "|" + codeAndId[1] + "|" + start;
        System.out.print("startRowkey" + startRowkey);
        String endRowkey = codeAndId[0] + "|" + codeAndId[1] + "|" + end;
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRowkey));//范围查询
        scan.setStopRow(Bytes.toBytes(endRowkey));
        scan.setCacheBlocks(false);
        //scan.setCaching(1000);
        List<PointData> list = HBaseUtils.genData(table, scan, pointName);
        table.close();
        return list;
    }



}


